package io.gitee.hefren.hhmq.client;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import io.gitee.hefren.hhmq.model.HhMessage;
import io.gitee.hefren.hhmq.model.Result;
import io.gitee.hefren.utils.http.HttpInvoker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Date 2024/7/21
 * @Author lifei
 */
public class HhBroker {

    private static final Logger log = LoggerFactory.getLogger(HhBroker.class);
    private static final String brokerURL = "http://localhost:8765";
    private static final String SEND = "send";
    private static final String SUBSCRIBE = "subscribe";
    private static final String UNSUBSCRIBE = "unsubscribe";
    private static final String RECEIVE = "receive";
    private static final String BATCHRECEIVE = "batchReceive";
    private static final String ACK = "ack";
    private static final String EMPTY = "";

    private static final Map<String, HhConsumer> consumers = Maps.newHashMap();
    private static final Gson gson = new Gson();

    static {
        init();
    }

    private static void init() {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleWithFixedDelay(HhBroker::listener, 1000, 1000, TimeUnit.MILLISECONDS);
    }

    private static void listener() {
        try {
            consumers.values().parallelStream().forEach(consumer -> {
                HhMessage<?> message = consumer.receive();
                if (Objects.isNull(message)) {
                    return;
                }
                consumer.getHhMessageListener().onMessage(message);
                consumer.ack(message);
            });
        }catch (Exception e) {
            log.error("===> listener message error: {}", e.getMessage());
        }
    }

    private HhBroker(){}

    private static class HhBrokerHolder {
        private static final HhBroker INSTANCE = new HhBroker();
    }

    public static HhBroker instance() {
        return HhBrokerHolder.INSTANCE;
    }

    /**
     * 创建生产者
     * @return
     */
    public HhProducer createProducer() {
        return new HhProducer(this);
    }

    /**
     * 创建消息者，并订阅主题
     * @param topic
     * @return
     */
    public HhConsumer createConsumerAndSubscribeTopic(String topic) {
        HhConsumer hhConsumer = new HhConsumer(this);
        hhConsumer.subscribe(topic);
        return hhConsumer;
    }

    /**
     * 发送消息
     * @param topic
     * @param hhMessage
     * @return
     */
    public boolean send(String topic, HhMessage<?> hhMessage) {
        String url = sendURL(topic);
        TypeToken<Result<Integer>> typeToken = new TypeToken<>(){};
        Result<Integer> result = HttpInvoker.httpPost(url, gson.toJson(hhMessage), typeToken);
        return result.isSuccess();
    }

    /**
     * 客户端订阅主题
     * @param topic
     * @param consumerId
     * @return
     */
    public boolean subscribe(String topic, String consumerId) {
        String url = subscribeURL(topic, consumerId);
        TypeToken<Result<String>> typeToken = new TypeToken<>(){};
        Result<String> result = HttpInvoker.httpPost(url, EMPTY, typeToken);
        return result.isSuccess();
    }

    /**
     * 客户端反订阅
     * @param topic
     * @param consumerId
     * @return
     */
    public boolean unsubscribe(String topic, String consumerId) {
        String url = unsubscribeURL(topic, consumerId);
        TypeToken<Result<String>> typeToken = new TypeToken<>(){};
        Result<String> result = HttpInvoker.httpPost(url, EMPTY, typeToken);
        return result.isSuccess();
    }


    /**
     * 客户端拉取消息
     * @param topic
     * @param consumerId
     * @return
     */
    public HhMessage<?> receive(String topic, String consumerId) {
        String url = receiveURL(topic, consumerId);
        TypeToken<Result<HhMessage<?>>> typeToken = new TypeToken<>(){};
        Result<HhMessage<?>> result = HttpInvoker.httpPost(url, EMPTY, typeToken);
        return result.getData();
    }

    /**
     * 批次获取消息
     * @param topic
     * @param consumerId
     * @param size
     * @return
     */
    public List<HhMessage<?>> batchReceive(String topic, String consumerId, Integer size) {
        String url = batchReceiveURL(topic, consumerId, size);
        TypeToken<Result<List<HhMessage<?>>>> typeToken = new TypeToken<>(){};
        Result<List<HhMessage<?>>> result = HttpInvoker.httpPost(url, EMPTY, typeToken);
        return result.getData();
    }

    /**
     * 确认消费者消费的偏移量
     * @param topic
     * @param consumerId
     * @param offset
     * @return
     */
    public boolean ack(String topic, String consumerId, Integer offset) {
        String url = ackURL(topic, consumerId, offset);
        TypeToken<Result<Integer>> typeToken = new TypeToken<>(){};
        Result<Integer> result = HttpInvoker.httpPost(url, EMPTY, typeToken);
        return result.isSuccess();
    }




    public boolean addConsumer(HhConsumer hhConsumer) {
        consumers.put(hhConsumer.getId(), hhConsumer);
        return true;
    }

    private String batchReceiveURL(String topic, String consumerId, Integer size) {
        return createUrl(BATCHRECEIVE, topic, consumerId, null, size);
    }

    private String ackURL(String topic, String consumerId, Integer offset) {
        return createUrl(ACK, topic, consumerId, offset, null);
    }

    private String receiveURL(String topic, String consumerId) {
        return createUrl(RECEIVE, topic, consumerId, null, null);
    }

    private String unsubscribeURL(String topic, String consumerId) {
        return createUrl(UNSUBSCRIBE, topic, consumerId, null, null);
    }

    private String subscribeURL(String topic, String consumerId) {
        return createUrl(SUBSCRIBE, topic, consumerId, null, null);
    }

    private String sendURL(String topic) {
        return createUrl(SEND, topic, null, null, null);
    }

    private String createUrl(String subPath, String topic, String consumerId, Integer offset, Integer size) {
        StringBuilder result = new StringBuilder();
        result.append(Strings.lenientFormat("%s/hhmq/%s?topic=%s", brokerURL, subPath, topic));
        if (Objects.nonNull(consumerId) && !consumerId.isBlank()) {
            result.append(Strings.lenientFormat("&consumerId=%s", consumerId));
        }
        if (Objects.nonNull(offset)) {
            result.append(Strings.lenientFormat("&offset=%s", offset));
        }
        if (Objects.nonNull(size)) {
            result.append(Strings.lenientFormat("&size=%s", size));
        }
        return result.toString();
    }
}
